package com.mask.im.sdk.config;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.Properties;

/**
 * Kafka自动配置类
 * 负责创建KafkaProducer并注册到Spring容器
 * @author jian
 */
@Configuration
@EnableConfigurationProperties(KafkaSdkProperties.class)
public class KafkaAutoConfiguration {

    private final KafkaSdkProperties kafkaSdkProperties;

    // 注入Kafka配置属性
    public KafkaAutoConfiguration(KafkaSdkProperties kafkaProperties) {
        this.kafkaSdkProperties = kafkaProperties;
    }

    /**
     * 将KafkaProducer注册为Spring Bean，由Spring管理生命周期
     * destroyMethod = "close" 确保容器销毁时自动关闭生产者
     */
    @Bean(destroyMethod = "close")
    public KafkaProducer<String, String> kafkaProducer() {
        Properties props = new Properties();
        props.put("bootstrap.servers", kafkaSdkProperties.getBootstrapServers());
        props.put("key.serializer", StringSerializer.class.getName());
        props.put("value.serializer", StringSerializer.class.getName());

        // 可靠性配置
        props.put("retries", kafkaSdkProperties.getRetries());
        props.put("acks", kafkaSdkProperties.getAcks());
        props.put("delivery.timeout.ms", kafkaSdkProperties.getDeliveryTimeoutMs());
        // 性能配置
        props.put("batch.size", kafkaSdkProperties.getBatchSize());
//        props.put("linger.ms", kafkaSdkProperties.getLingerMs());
        props.put("buffer.memory", kafkaSdkProperties.getBufferMemory());
        props.put("auto.create.topics.enable", true);
        return new KafkaProducer<>(props);
    }
}
